﻿using System;

using Confluent.Kafka;
using Microsoft.Extensions.Logging;
using Unity.Commons;
using Unity.Connections;
using Unity.Credentials;
using Unity.Logger;

namespace Unity.Kafka
{
    public class UnityKafkaConsumer : IUnityConsumer
    {
        private static readonly ILogger Logger = UnityLogger.ConsoleLoggerFactory.CreateLogger(nameof(UnityKafkaConsumer));
        
        private readonly UnityKafkaCredentials _credentials;
        private IConsumer<Ignore, byte[]> _consumer;

        public UnityKafkaConsumer(UnityKafkaCredentials credentials)
        {
            _credentials = credentials;
        }

        public bool Connect(IUnityCredentials credentials)
        {
            try
            {
                _consumer = _credentials.GetConsumer();
                _consumer.Subscribe(_credentials.Topic);
                return true;
            }
            catch(Exception exception)
            {
                Logger.LogError("Error connect to Kafka consumer. Exception: {}", exception.Message);
                return false;
            }
        }

        public void Close()
        {
            if(_consumer != null)
            {
                _consumer.Close();
                _consumer.Dispose();
            }
        }

        public void Dispose()
        {
            Close();
        }

        public byte[] Pull(int timeout)
        {
            try
            {
                var consumeResult = _consumer.Consume(timeout);
                if (consumeResult != null)
                {
                    return consumeResult.Message.Value;
                }
            }
            catch (Exception exception)
            {
                Logger.LogError("Error pull to Kafka consumer. Exception: {}", exception.Message);
            }
            return null;
        }
    }
}
